@file:OptIn(ExperimentalTime::class)

package com.gitee.wsl.func.filter.bloomfilter

import com.gitee.wsl.func.Supplier
import com.gitee.wsl.func.consumer.Consumer
import com.gitee.wsl.func.filter.bloomfilter.DefaultBloomFilterTest.SparseDefaultBloomFilter
import com.gitee.wsl.func.filter.bloomfilter.TestingHashers.populateFromHashersFrom1AndFrom11
import com.gitee.wsl.func.filter.bloomfilter.api.BitMapExtractor
import com.gitee.wsl.func.filter.bloomfilter.api.BloomFilter
import com.gitee.wsl.func.filter.bloomfilter.api.Hasher
import com.gitee.wsl.func.filter.bloomfilter.api.IndexExtractor
import com.gitee.wsl.func.filter.bloomfilter.bean.Shape
import com.gitee.wsl.func.filter.bloomfilter.extractor.LayerManager
import com.gitee.wsl.func.filter.bloomfilter.extractor.LayerManager.Cleanup
import com.gitee.wsl.func.filter.bloomfilter.extractor.LayerManager.ExtendCheck
import com.gitee.wsl.func.predicate.Predicate
import com.gitee.wsl.text.format.format
import com.gitee.wsl.time.between
import com.gitee.wsl.time.isAfter
import com.gitee.wsl.time.isBefore
import kotlin.time.Clock
import kotlin.time.Instant
import kotlin.jvm.JvmOverloads
import kotlin.test.Test
import kotlin.test.assertContentEquals
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertTrue
import kotlin.time.Duration
import kotlin.time.ExperimentalTime


class LayeredBloomFilterTest : AbstractBloomFilterTest<LayeredBloomFilter<SimpleBloomFilter>>() {
    /**
     * A Predicate that advances after a quantum of time.
     */
    internal class AdvanceOnTimeQuanta<T : BloomFilter<T>>(var quanta: Duration) : Predicate<LayerManager<TimestampedBloomFilter<T>>> {
        override fun test(layerManager: LayerManager<TimestampedBloomFilter<T>>): Boolean {
            // cannot use getTarget() as it causes recursion.
            return layerManager.last().timestamp.plus(quanta).isBefore(Clock.System.now())
        }
    }

    /**
     * A Consumer that cleans the list based on how long each filters has been in
     * the list.
     */

    internal class CleanByTime<T : TimestampedBloomFilter<*>>(var elapsedTime: Duration) : Consumer<MutableList<T>> {
        override fun accept(t: MutableList<T>) {
            val min = Clock.System.now().minus(elapsedTime)
            val iter = t.iterator()
            while (iter.hasNext()) {
                val bf: TimestampedBloomFilter<*> = iter.next()!!
                if (bf.timestamp.isAfter(min) || bf.timestamp == min) {
                    return
                }
                dbgInstrument.add(
                   String.format(
                        "Removing old entry: T:%s (Aged: %s) \n",
                        bf.timestamp, Duration.between(bf.timestamp, min)
                    )
                )
                iter.remove()
            }
        }
    }

    internal class NumberedBloomFilter(
        shape: Shape,
        var value: Int,
        var sequence: Int
    ) : WrappedBloomFilter<NumberedBloomFilter, SimpleBloomFilter>(SimpleBloomFilter(shape)) {
        override fun copy(): NumberedBloomFilter {
            return NumberedBloomFilter(shape, value, sequence)
        }
    }

    /**
     * A Bloom filter implementation that tracks the creation time.
     */
    class TimestampedBloomFilter<T : BloomFilter<T>> @JvmOverloads internal constructor(
        bf: T,
        val timestamp: Instant = Clock.System.now(),
    ) : WrappedBloomFilter<TimestampedBloomFilter<T>, T>(bf) {
        override fun copy(): TimestampedBloomFilter<T> {
            return TimestampedBloomFilter(wrapped.copy(), timestamp)
        }
    }

    // instrumentation to record timestamps in dbgInstrument list
    private val dbg: Predicate<BloomFilter<*>> = Predicate { bf ->
        val tbf = bf as TimestampedBloomFilter<*>
        val ts = Clock.System.now()
        dbgInstrument.add(
            String.format(
                "T:%s (Elapsed:%s)- EstN:%s (Card:%s)\n", tbf.timestamp, Duration.between(tbf.timestamp, ts),
                tbf.estimateN(), tbf.cardinality()
            )
        )
        true
    }

    // *** end of instrumentation ***
    protected override fun createEmptyFilter(shape: Shape): LayeredBloomFilter<SimpleBloomFilter> {
        return fixed(shape, 10)
    }

    protected fun makeFilter(h: Hasher): BloomFilter<*> {
        val bf = SparseBloomFilter(testShape)
        bf.merge(h)
        return bf
    }

    protected fun makeFilter(p: IndexExtractor): BloomFilter<*> {
        val bf = SparseBloomFilter(testShape)
        bf.merge(p)
        return bf
    }

    protected fun makeFilter(vararg values: Int): BloomFilter<*> {
        return makeFilter(IndexExtractor.fromIndexArray(*values))
    }

    private fun setupFindTest(): LayeredBloomFilter<SimpleBloomFilter> {
        val filter = fixed(testShape, 10)
        filter.merge(TestingHashers.FROM1)
        filter.merge(TestingHashers.FROM11)
        filter.merge(IncrementingHasher(11, 2))
        filter.merge(populateFromHashersFrom1AndFrom11(SimpleBloomFilter(testShape)))
        return filter
    }

//    @Test
//    public override fun testCardinalityAndIsEmpty() {
//        val layerManager = LayerManager.builder<SimpleBloomFilter>().setExtendCheck(ExtendCheck.neverAdvance<T>())
//                .setSupplier({ SimpleBloomFilter(testShape) }).get()
//        testCardinalityAndIsEmpty(LayeredBloomFilter<SimpleBloomFilter>(testShape, layerManager))
//    }

    // ***** TESTS THAT CHECK LAYERED PROCESSING ******
//    @Test
//    fun testCleanup() {
//        val sequence = intArrayOf(1)
//        val layerManager: LayerManager<NumberedBloomFilter> =
//            LayerManager.<NumberedBloomFilter> builder <LayeredBloomFilterTest.NumberedBloomFilter>()
//                .setSupplier({ NumberedBloomFilter(testShape, 3, sequence[0]++) })
//                .setExtendCheck(ExtendCheck.neverAdvance<T?>())
//                .setCleanup({ ll -> ll.removeIf({ f -> (f.value-- === 0) }) }).get()
//        val underTest = LayeredBloomFilter<NumberedBloomFilter>(testShape, layerManager)
//        assertEquals(1, underTest.depth)
//        underTest.merge(randomHasher())
//        underTest.cleanup() // first count == 2
//        assertEquals(1, underTest.depth)
//        underTest.next() // first count == 1
//        assertEquals(2, underTest.depth)
//        underTest.merge(randomHasher())
//        underTest.cleanup() // first count == 0
//        var f = underTest.get(0)
//        assertEquals(1, f!!.sequence)
//
//        assertEquals(2, underTest.depth)
//        underTest.cleanup() // should be removed ; second is now 1st with value 1
//        assertEquals(1, underTest.depth)
//        f = underTest.get(0)
//        assertEquals(2, f!!.sequence)
//
//        underTest.cleanup() // first count == 0
//        underTest.cleanup() // should be removed.  But there is always at least one
//        assertEquals(1, underTest.depth)
//        f = underTest.get(0)
//        assertEquals(3, f!!.sequence) // it is a new one.
//    }

    /**
     * Tests that the estimated union calculations are correct.
     */
    @Test
    fun testEstimateUnionCrossTypes() {
        val bf= createFilter(testShape, TestingHashers.FROM1)
        val bf2 = SparseDefaultBloomFilter(testShape)
        bf2.merge(TestingHashers.FROM11)

        assertEquals(2, bf.estimateUnion(bf2))
        assertEquals(2, bf2.estimateUnion(bf))
    }

//    @Test
//    @Throws(InterruptedException::class)
//    fun testExpiration() {
//        // this test uses the instrumentation noted above to track changes for debugging
//        // purposes.
//
//        // list of timestamps that are expected to be expired.
//
//        val lst: List<Instant?> = ArrayList()
//        val shape = Shape.fromNM(4, 64)
//
//        // create a filter that removes filters that are 4 seconds old
//        // and quantises time to 1 second intervals.
//        val underTest = createTimedLayeredFilter(
//            shape, Duration.ofMillis(600),
//            Duration.ofMillis(150)
//        )
//
//        for (i in 0..9) {
//            underTest.merge(randomHasher())
//        }
//        underTest.processBloomFilters(dbg.and(Predicate { x: BloomFilter? -> lst.add((x as TimestampedBloomFilter<*>).timestamp) }))
//        assertTrue(underTest.depth > 1)
//
//        //java.lang.Thread.sleep(300)
//        for (i in 0..9) {
//            underTest.merge(randomHasher())
//        }
//        dbgInstrument.add("=== AFTER 300 milliseconds ====\n")
//        underTest.processBloomFilters(dbg)
//
//        //java.lang.Thread.sleep(150)
//        for (i in 0..9) {
//            underTest.merge(randomHasher())
//        }
//        dbgInstrument.add("=== AFTER 450 milliseconds ====\n")
//        underTest.processBloomFilters(dbg)
//
//        // sleep 200 milliseconds to ensure we cross the 600 millisecond boundary
//        //java.lang.Thread.sleep(200)
//        underTest.merge(randomHasher())
//        dbgInstrument.add("=== AFTER 600 milliseconds ====\n")
//        assertTrue(
//            underTest.processBloomFilters(dbg.and(Predicate { x: BloomFilter? -> !lst.contains((x as TimestampedBloomFilter<*>).timestamp) })),
//            "Found filter that should have been deleted: " + dbgInstrument.get(dbgInstrument.size() - 1)
//        )
//    }

    @Test
    fun testFindBitMapExtractor() {
        val filter = setupFindTest()

        var indexExtractor = TestingHashers.FROM1.indices(testShape)
        var bitMapExtractor = BitMapExtractor.fromIndexExtractor(indexExtractor, testShape.numberOfBits)

        var expected = intArrayOf(0, 3)
        var result = filter.find(bitMapExtractor)
        assertContentEquals(expected, result)

        expected = intArrayOf(1, 3)
        indexExtractor = TestingHashers.FROM11.indices(testShape)
        bitMapExtractor = BitMapExtractor.fromIndexExtractor(indexExtractor, testShape.numberOfBits)
        result = filter.find(bitMapExtractor)
        assertContentEquals(expected, result)
    }

    @Test
    fun testFindBloomFilter() {
        val filter = setupFindTest()
        var expected = intArrayOf(0, 3)
        var result = filter.find(TestingHashers.FROM1)
        assertContentEquals(expected, result)
        expected = intArrayOf(1, 3)
        result = filter.find(TestingHashers.FROM11)
        assertContentEquals(expected, result)
    }

    @Test
    fun testFindIndexExtractor() {
        var indexExtractor = TestingHashers.FROM1.indices(testShape)
        val filter = setupFindTest()

        var expected = intArrayOf(0, 3)
        var result = filter.find(indexExtractor)
        assertContentEquals(expected, result)

        expected = intArrayOf(1, 3)
        indexExtractor = TestingHashers.FROM11.indices(testShape)
        result = filter.find(indexExtractor)
        assertContentEquals(expected, result)
    }

    @Test
    fun testGetLayer() {
        val bf = SimpleBloomFilter(testShape)
        bf.merge(TestingHashers.FROM11)
        val filter = fixed(testShape, 10)
        filter.merge(TestingHashers.FROM1)
        filter.merge(TestingHashers.FROM11)
        filter.merge(IncrementingHasher(11, 2))
        filter.merge(populateFromHashersFrom1AndFrom11<SimpleBloomFilter>(SimpleBloomFilter(testShape)))
        assertContentEquals(bf.asBitMapArray(), filter.get(1)!!.asBitMapArray())
    }

    @Test
    fun testMultipleFilters() {
        val filter = fixed(testShape, 10)
        filter.merge(TestingHashers.FROM1)
        filter.merge(TestingHashers.FROM11)
        assertEquals(2, filter.depth)
        assertTrue(filter.contains(makeFilter(TestingHashers.FROM1)))
        assertTrue(filter.contains(makeFilter(TestingHashers.FROM11)))
        val t1 = makeFilter(6, 7, 17, 18, 19)
        assertFalse(filter.contains(t1))
        assertFalse(filter.copy().contains(t1))
        assertTrue(filter.flatten().contains(t1))
    }

    @Test
    fun testNext() {
        val layerManager = LayerManager.builder<SimpleBloomFilter>().setSupplier( Supplier { SimpleBloomFilter(testShape) }).get()
        val filter = LayeredBloomFilter(testShape, layerManager)
        filter.merge(TestingHashers.FROM1)
        filter.merge(TestingHashers.FROM11)
        assertEquals(1, filter.depth)
        filter.next()
        filter.merge(IncrementingHasher(11, 2))
        assertEquals(2, filter.depth)
        assertTrue(filter.get(0)!!.contains(TestingHashers.FROM1))
        assertTrue(filter.get(0)!!.contains(TestingHashers.FROM11))
        assertFalse(filter.get(0)!!.contains(IncrementingHasher(11, 2)))
        assertFalse(filter.get(1)!!.contains(TestingHashers.FROM1))
        assertFalse(filter.get(1)!!.contains(TestingHashers.FROM11))
        assertTrue(filter.get(1)!!.contains(IncrementingHasher(11, 2)))
    }

    companion object {
        // ***example of instrumentation ***
        private val dbgInstrument = ArrayList<String>()

        /**
         * Creates a LayeredBloomFilter that retains enclosed filters for
         * `duration` and limits the contents of each enclosed filter to a time
         * `quanta`. This filter uses the timestamped Bloom filter internally.
         *
         * @param shape    The shape of the Bloom filters.
         * @param duration The length of time to keep filters in the list.
         * @param quanta   The quantization factor for each filter. Individual filters
         * will span at most this much time.
         * @return LayeredBloomFilter with the above properties.
         */
//        fun createTimedLayeredFilter(
//            shape: Shape,
//            duration: Duration,
//            quanta: Duration,
//        ): LayeredBloomFilter<TimestampedBloomFilter<SimpleBloomFilter>> {
//            val builder = LayerManager.builder()
//            val cleanup =
//                Cleanup.removeEmptyTarget<T>().andThen<Any>(LayeredBloomFilterTest.CleanByTime<Any?>(duration))
//            val layerManager = builder
//                .setSupplier(Supplier { LayeredBloomFilterTest.TimestampedBloomFilter<T?>(SimpleBloomFilter(shape)) })
//                .setCleanup(cleanup)
//                .setExtendCheck(
//                    LayeredBloomFilterTest.AdvanceOnTimeQuanta<Any?>(quanta)
//                        .or(ExtendCheck.advanceOnSaturation<T?>(shape.estimateMaxN()))
//                )
//                .get()
//            return LayeredBloomFilter<T?>(shape, layerManager)
//        }

        /**
         * Creates a fixed size layered bloom filter that adds new filters to the list,
         * but never merges them. List will never exceed maxDepth. As additional filters
         * are added earlier filters are removed.  Uses SimpleBloomFilters.
         *
         * @param shape    The shape for the enclosed Bloom filters.
         * @param maxDepth The maximum depth of layers.
         * @return An empty layered Bloom filter of the specified shape and depth.
         */
        fun fixed(shape: Shape, maxDepth: Int): LayeredBloomFilter<SimpleBloomFilter> {
            return fixed(shape, maxDepth, Supplier { SimpleBloomFilter(shape) })
        }

        /**
         * Creates a fixed size layered bloom filter that adds new filters to the list,
         * but never merges them. List will never exceed maxDepth. As additional filters
         * are added earlier filters are removed.
         *
         * @param shape    The shape for the enclosed Bloom filters.
         * @param maxDepth The maximum depth of layers.
         * @param supplier A supplier of the Bloom filters to create layers with.
         * @return An empty layered Bloom filter of the specified shape and depth.
         */
        fun <T : BloomFilter<T>> fixed(shape: Shape, maxDepth: Int, supplier: Supplier<T>): LayeredBloomFilter<T> {
            val builder = LayerManager.builder<T>()
            builder.setExtendCheck(ExtendCheck.advanceOnPopulated<T>())
                .setCleanup(Cleanup.onMaxSize<T>(maxDepth)).setSupplier(supplier)
            return LayeredBloomFilter<T>(shape, builder.get())
        }
    }
}